package GroupWare;

import java.io.BufferedInputStream;
import java.io.ByteArrayInputStream;
import java.io.DataInputStream;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.PrintWriter;
import java.io.RandomAccessFile;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.util.Date;
import java.util.LinkedList;
import java.util.Timer;
import java.util.TimerTask;
import java.util.Vector;
import java.util.concurrent.ConcurrentHashMap;

import DBUtil.CNN;
import InetAgreement.InetDataEncapsulation;
import Tools.Log;
import Tools.Units;
import Windows.MainWindows;
import XML.XML;
import XML.XMLNode;

/**
 * 下载任务类
 * 
 * @author Administrator
 * 
 */
public class DownloadTask extends ShareFile implements Runnable {
	/**
	 * 任务种子类
	 * 
	 * @author Administrator
	 * 
	 */
	public class PipThread extends Pip implements Runnable {
		public boolean over;
		private Thread thread;
		private Socket sok;// 客户端
		private long offset;// 当前字节偏移位置
		private long end;// 文件字节传输结束位置
		private PrintWriter dos;// 数据发送流
		private InputStream dis;// 数据读取流
		private volatile boolean Stop = true;// 任务中是否出现异常,初始为真
		private volatile long conConsume = 0;// 建立连接耗费时间
		private volatile long transmitBT = 0;// 已传送字节数
		private volatile long transmitTime = 0;// 已传送时间
		private volatile int speed = -1;// -1 表示刚刚连接
		private volatile LinkedList<ParcelBytes> buf3;
		byte[] buf = new byte[ParcelBytes.getBuffer()];// 2K缓冲
		ParcelBytes buf2 = null;// 二级缓冲区
		
		
		public PipThread(LinkedList<ParcelBytes> albt) {
			this.buf3 = albt;
		}

		public PipThread(Pip p, LinkedList<ParcelBytes> albt) {
			this(albt);
			this.SetPip(p);// 设置种子
			// 连接用户,并测试连接速度
			Date t = new Date();
			try {
				if (p.getAddress() != null) {
					Connect();
				}
				// 连接成功，加入待命区
				alwait.add(this);
				this.conConsume = new Date().getTime() - t.getTime();
				Log.print("[有效种子]" + p + "\tConectionTime:" + conConsume);
			} catch (IOException ie) {
				Log.print("[无效种子]" + p);
				ie.printStackTrace();
			}
		}

		/**
		 * 用来保持连接 多个线程调用此方法会同步锁定，不依赖于对象
		 * 
		 * @throws IOException
		 */
		public synchronized void Connect() throws IOException {
			if (getAddress() == null)
				return;
			if (sok == null || !sok.isConnected() || sok.isClosed()) {
				try {
					sok = new Socket();
					sok.connect(new InetSocketAddress(getAddress(), getPort()), 3000);
					sok.setSoTimeout(30000);// 设置超时时间30秒
					// 创建网络输出输入流
					dos = new PrintWriter(sok.getOutputStream(), true);
					dis = new DataInputStream(new BufferedInputStream(sok.getInputStream()));
					Log.print("[连接种子]" + super.toString() + "  连接成功");
				} catch (IOException e) {
					setAddress(null);
					setPort(0);
					sok.close();
					sok=null;
					Log.print("[连接种子]" + super.toString() + "  连接失败，改服务器转发");
				}
			}
		}

		public LinkedList<ParcelBytes> getAlbt() {
			return buf3;
		}

		public ParcelBytes getBuf2() {
			return buf2;
		}

		public long getEnd() {
			return end;
		}

		public long getOffset() {
			return offset;
		}

		/**
		 * 取通讯网速
		 * 
		 * @return 速率单位 byte/秒 0表示没有传输
		 */
		public long getSpeedNet() {
			long bt = transmitBT;
			long t = transmitTime;
			// if(t>10)transmitBT=transmitBT=0;//10秒重置
			if (t <= 0)
				return 0;
			long speed = bt / t * 1000;
			return speed;
		}

		public long getTransmitBT() {
			return transmitBT;
		}

		public long getTransmitTime() {
			return transmitTime;
		}

		/**
		 * 种子传输线程方法
		 */
		@Override
		public void run() {
			Log.print("[PipThread]Startting..." + this);
			// //////////////////////////////////
			Stop = false;
			// alwait.remove(this);
			// //////////////////////////////////
			try {
				Connect();// 保证连接
				speed = speed == -1 ? 0 : speed;// 取消连接标记，进入传输
			} catch (IOException e) {
				this.Stop = true;// 停止并等待继承任务
				Log.print("[废弃种子]" + e.getMessage());
				return;
			}

			try {
				XMLNode xf = new XMLNode(getAddress()!=null&&sok!=null&&sok.isConnected() ? "get" : "dloadGet", true);
				if (sok == null) {
					xf.put("tgcode", getGcode());
					xf.put("tgucode", getGucode());
				}
				xf.put("userid", getUserID());
				xf.put("md5", md5);

				do {
					long start = System.currentTimeMillis();// 开始计时
					xf.put("offset", String.valueOf(offset));
					xf.put("end", String.valueOf(end));
					// ///////////////////////////////////////////////
					int i, k;// 已缓存数量
					long count = end - offset;// 需要读取的长度

					// 发送传输指令
					send(xf);

					if (buf2 == null){
						buf2 = ParcelBytes.CreateParcelBytes(offset);// 继续中断任务，不需要重新创建二级缓冲区
					}
					// ///////////////////////////////////////////////
					// System.out.println("[Ready]offset:"+offset+"\tend:"+end+"\tcount:"+count+"\tbuf:"+buf.length);
					if (dis == null) {
						throw new IOException("Recv file data timeout!");
					}
					Log.print("write buf2 "+dis.available());
					while (count > 0 && (i = dis.read(buf)) > 0) {
						i = i <= count ? i : (int) count;// 实际读取的字节数多余欲读取数量，校正读取数量

						k = buf2.write(buf, 0, i);// 输出到二级缓存

						// 检查二级缓存写入溢出异常
						if (k < i) {
							throw new IOException("[PipThread]Exception, ParcelBytes byte out!!!");
						}

						// System.out.println("[read]offset:"+offset+"\tend:"+end+"\tcount:"+count+"\ti:
						// "+i+"\tk: "+k);
						// 统计处理
						count -= i;// 统计剩余数量
						offset += i;// 校准指针偏移位置
						transmitBT += i;// 累计流量
					}
					Log.print("write to buf2 over");
					synchronized (buf3) {// 保护公共对象
						buf3.add(buf2);// 添加到缓存输出区，等待缓存输出线程写到文件
						 Log.print("[PipThread]buf2 add to bufferList, \t size:"+buf2.getSize()+"  count: "+buf3.size());
						over = true;
						buf2 = null;// 清空，下次自动获取新的
					}
					transmitTime += System.currentTimeMillis() - start;// 累计消耗时间
				} while (doing && assignTask(this));// 继续分配任务
				Log.print("pip task over");
				// if (buf2 != null)
				// synchronized (buf3) {
				// buf3.add(buf2);// 添加到缓存输出区，等待缓存输出线程写到文件
				// // Log.print("[PipThread]residual buf2,
				// // "+this+"\t"+buf2);
				// buf2 = null;// 清空，下次自动获取新的
				// }
				// 回到待命池，只有完成任务的种子才需要返回到待命区
				synchronized (DownloadTask.this) {
					altask.remove(this);
					if (alwait.indexOf(this) == -1){
						alwait.add(this);
					}
				}
			} catch (Exception e) {// 中途异常任务将遗留在任务区，等待其他种子继承
				e.printStackTrace();
				Log.print("种子线程:异常！>>" + e.getMessage() + "\tTASK:" + this.toString());
				this.setPort(0);// 标记此种子作废
				try {
					if (dis != null)
						dis.close();
					if (dos != null)
						dos.close();
					if (sok != null)
						sok.close();// 关闭套接字
				} catch (Exception e1) {
				}
				infomEvent(EVENT_EXCEPTION);// 触发异常事件
			} finally {
				this.Stop = true;
				thread = null;
				Log.print("[PipThread]end..." + this);
			}
		}

		public void send(XMLNode msg) {
			if (getAddress() != null && sok != null && sok.isConnected()) {
				dos.println(dp.Package(msg.toString()));
				dos.flush();
			} else {
				String md5 = msg.get("md5");
				String offset = msg.get("offset");
				String end = msg.get("end");
				dis = null;
				MainWindows.getInstance().client.send(msg);
				int count = 0;
				String key = md5 + ":" + offset + "-" + end;
				int recvTimeout = Integer.parseInt(CNN.getConfig().GetPathValue("Transfers:RecvTimeout").toString());
				while (dis == null && count++ < recvTimeout*1000) {
					try {
						if (dataRecv.containsKey(key)) {
							byte[] data = dataRecv.remove(key);
							if (data != null) {
								Log.print("取到种子数据：" + key + ": " + data.length);
								dis = new ByteArrayInputStream(data);
							}
						}
						Thread.sleep(1);
					} catch (Exception e) {
						e.printStackTrace();
					}
				}
			}
		}

//		public void setAlbt(LinkedList<ParcelBytes> albt) {
//			this.buf3 = albt;
//		}

		public void setBuf2(ParcelBytes buf2) {
			this.buf2 = buf2;
		}

		public void setEnd(long end) {
			this.end = end;
		}

		public void setOffset(long offset) {
			this.offset = offset;
		}

		public void setTransmitBT(long transmitBT) {
			this.transmitBT = transmitBT;
		}

		public void setTransmitTime(long transmitTime) {
			this.transmitTime = transmitTime;
		}

		// -- 创建种子任务线程
		public void start() {
			if (thread == null) {
				thread = new Thread(this);
				thread.start();
			}
		}

		/**
		 * 接管其他任务
		 * 
		 * @param fss
		 */
		public void takeOverOf(PipThread fss) {
			this.offset = fss.getOffset();
			this.end = fss.getEnd();
			this.transmitBT = fss.getTransmitBT();
			this.transmitTime = fss.getTransmitTime();
			this.buf2 = fss.getBuf2();

		}

		@Override
		public String toString() {
			return "[PipThread:offset=" + offset + "\tend=" + end + "\t" + super.toString() + "]";
		}
	}
	/**
	 * 输出最近的缓存片
	 */
	public static final int LATELY = 0;
	/**
	 * 接着上次输出的偏移位置
	 */
	public static final int TEAM_NEXT = 1;
	public static final int EVENT_ORVER = 1;
	public static final int EVENT_PERIODS = 2;
	public static final int EVENT_EXCEPTION = 3;
	public static final int EVENT_PAUSE = 0;
	public static int ThreadSum = 60;
	private volatile XML xml = new XML();
	private volatile long achieve = 0;// 完成写入字节数
	private volatile int threadSum = ThreadSum;// 并发线程数
	private volatile long rafOffset = 0;// 缓存输出偏移量，链式输出。
	private volatile long alEnd = 0;// 活动任务队列最后成员的文件终点；分配种子任务assignTask(FileSocket)
	private volatile long end = 0;// 文件缓存输出偏移量
	private volatile File XMLFile;
	private volatile long transmitTime = 1;
	private Thread output;// 缓存输出线程
	private volatile LinkedList<DownloadTaskListener> listener = new LinkedList<DownloadTaskListener>();// 任务监听器

	private static int wise = LATELY;// 缓存输出方式

	static {
		try {
			wise = Integer.parseInt((String) CNN.getConfig().GetPathValue("Transfers.Download.writetype"));
			ThreadSum = Integer.parseInt((String) CNN.getConfig().GetPathValue("Transfers.Download.threadsum"));
			Log.print("[DownLoadTask]readConfig, Bytes Out wise:" + wise + "\tMaxThreadSum:" + ThreadSum);
		} catch (Exception e) {
		}

	}

	private static final InetDataEncapsulation dp = new InetDataEncapsulation();// 数据封包对象

	public static int getWise() {
		return wise;
	}

	private volatile LinkedList<ParcelBytes> albt = new LinkedList<ParcelBytes>(); // 文件字节缓存集合

	private volatile LinkedList<PipThread> altask = new LinkedList<PipThread>(); // 执行任务集合

	private volatile LinkedList<PipThread> alwait = new LinkedList<PipThread>(); // 待命任务集合

	public File tmpfile;
	public RandomAccessFile raf;// 任务随机访问文件流

	private Timer timer; // 任务时钟

	private TimerTask tt; // 刷新显示的任务

	private volatile long Speed = 0; // 平均网速

	private volatile long LastModified;// 最后更新时间

	public volatile long length; // 文件长度

	private ShareRoot sp; // 共享目录

	private volatile boolean doing = false;

	public static ConcurrentHashMap<String, byte[]> dataRecv = new ConcurrentHashMap<String, byte[]>();

	/**
	 * 从配置文件读取任务记录信息
	 * 
	 * @param xml
	 *            续传任务配置
	 * @param sp
	 *            共享目录
	 */

	public DownloadTask(XML xml, ShareRoot sp) {
		this(xml.copy(xml), sp);
		super.md5 = xml.FLAG;
		Vector<XMLNode> al = new Vector<XMLNode>();
		xml.getList(md5, al, false);
		XMLFile = new File(CNN.taskDir.getAbsolutePath() + "/" + md5 + ".xml");
		while (al.size() > 0) {
			XMLNode xf = al.remove(0);
			String gcode = xf.get("gcode");
			String gucode = xf.get("gucode");
			String userid = xf.get("userid");
			String address = xf.get("address");
			long offset = Long.parseLong(xf.get("offset"));
			long end = Long.parseLong(xf.get("end"));
			PipThread fs;
			try {
				fs = new PipThread(albt);
				fs.SetPip(new Pip(gcode != null ? Long.parseLong(gcode) : 0, gucode != null ? Long.parseLong(gucode) : 0, userid, address != null ? InetAddress.getByName(address) : null, address != null ? Integer.parseInt(xf.get("port")) : 0));
				fs.setOffset(offset);
				fs.setEnd(end);
				fs.Stop = true;// 标记等待接力
				fs.over = Boolean.parseBoolean("over");
				if (xf.FLAG.equalsIgnoreCase("TASK")) {
					if (fs.over == false) {
						alwait.add(fs);// 按顺序列入任务区，等待接力
					}
					altask.add(fs);
				} else if (xf.FLAG.equalsIgnoreCase("WAIT")) {
					alwait.add(fs);// 按顺序列入任务区，等待接力
				}
			} catch (Exception e) {
				e.printStackTrace();
			}// 注意创建后自动列入待命区
		}
		Log.print("[DownloadTask]Build..." + this);
	}

	public DownloadTask(XMLNode xf, ShareRoot sp) {
		super(new File(xf.get("filepath")), xf.get("md5"));
		this.sp = sp; // 共享目录

		tmpfile = new File(this.getF().getAbsolutePath()+".tmp");
		this.XMLFile = new File(CNN.taskDir.getAbsolutePath() + "/" + md5 + ".xml");

		altask.clear();
		alwait.clear();

		xml.copy(xf); // 复制
		xml.setLeafage(false);

		length = Long.parseLong(xf.get("size"));
		LastModified = Long.parseLong(xf.get("time"));

		// //////////////读取配置文件//////////////////
		try {
			end = Long.parseLong(xf.get("@end")); // 读取最后写入位置
			alEnd = Long.parseLong(xf.get("@alEnd")); // 读取最后一个任务的目标偏移
			achieve = Long.parseLong(xf.get("@achieve")); // 读取已完成字节数量
			threadSum = Integer.parseInt(xf.get("@ThreadSum")); // 读取线程数量
			rafOffset = Long.parseLong(xf.get("@rafOffset")); // 读取写入位置
			transmitTime = Long.parseLong(xf.get("@transmitTime")); // 读取任务累计时程

		} catch (Exception e) {
		}
		// ///////////////////////////////////////////
	}

	public void addListener(DownloadTaskListener dtl) {
		if (listener.indexOf(dtl) == -1)
			listener.add(dtl);
	}

	/**
	 * 添加种子
	 * 
	 * @param p
	 *            种子
	 */
	public void addSeed(Pip p) {
		// 查找种子是否存在
		if (pipExist(p)) {
			Log.print("已存在的种子：" + p);
			return;
		}
		new PipThread(p, albt);

	}

	/**
	 * 自动从待命区分配任务
	 */
	private synchronized void allot() {
		if (doing && !isOK() && altask.size() < threadSum) {// 低于n个线程
			PipThread fs;
			// 优秀种子优先
			while ((fs = nextPip()) != null && altask.size() < threadSum) {
				if (assignTask(fs)) {// 判断任务分配是否成功
					fs.start();// 启动任务线程;// 分配任务
					Units.sleep(100);
				} else {
					alwait.add(fs);// 没有获得任务回到待命区
//					Log.print("[分配任务]没有可分配的任务," + fs);
					break;
				}
			}
		}
	}

	/**
	 * 为指定种子分配任务 注意：这里不启动线程，只是分配任务
	 * 
	 * @param fs
	 * @return
	 */
	private synchronized boolean assignTask(PipThread fs) {
		PipThread fss = null;
		if (fs.over == false && fs.offset != 0 && fs.end != 0) {
			return true;
		}
		// //////////////////查找中途异常任务，并接管它//////////////////////
		for (int i = 0; i < altask.size(); i++) {
			fss = altask.get(i);
			if (fs != fss && fss.Stop) {// 查找异常任务
				fs.takeOverOf(fss);// 接管任务
				fs.Stop = false;
				alwait.remove(fs);// 跳出待命区
				altask.remove(fss);// 销毁异常任务
				if (altask.indexOf(fs) == -1)
					altask.add(i, fs);// 进入任务区
				return true;
			}
		}
		// /////////////////////////接着最后一个任务继续////////////////////////////
		long start = alEnd;// 接着队列末尾
		long end = start + (fs.getAddress() != null ? ParcelBytes.getSecondBufferedSize() : 1024 * 32);// 给每个任务分配1M个任务
		end = end > length ? length : end;
		// System.out.println("！！！！！s:"+start+"\tend:"+end+"\tlength:"+length+"\tBufferedSize:"+getBufferedSize());
		if (start < length) {
			alEnd = end;// 更新任务队列末尾偏移量
			fs.setOffset(start);
			fs.setEnd(end);
			if (altask.indexOf(fs) == -1)
				altask.add(fs);// 追加在后面
			alwait.remove(fs);// 分配任务成功，从待命区移除
			return true;
		}

		return false;
	}

	public long getAchieve() {
		return achieve;
	}

	/**
	 * 取缓冲区字节总长度
	 * 
	 * @return
	 */
	public long getAmortizeLength() {
		long size = 0;
		for (int i = 0; i < albt.size(); i++) {
			size += albt.get(i).getBytes().length;
		}
		return size;
	}

	/**
	 * 文件并发传输速率，种子是并发传输的，因此整体传输速率是任务种子的速率和。
	 * 
	 * @return
	 */
	public long getAvgSpeed() {
		long count = 0;
		for (int i = 0; i < altask.size(); i++)
			count += altask.get(i).getSpeedNet();
		return count;
	}

	/**
	 * 完成度百分比
	 * 
	 * @return
	 */
	public int getCourse() {
		return (int) (100.0*achieve / length );
	}

	public long getLastModified() {
		return LastModified;
	}

	/**
	 * 取出下一个资源缓存块
	 * 
	 * @return
	 */
	public ParcelBytes getNextFB(int type) {
//		Log.print("getNextFB  "+type);
		// 初始化提取方案
		ParcelBytes fbt = type == TEAM_NEXT ? null : type == LATELY ?albt.size() > 0? albt.get(0) : null:null;
		synchronized (albt) {
			for (int i = 0; i < albt.size(); i++) {
				ParcelBytes tmp = albt.get(i);
				if (tmp != null) {
					switch (type) {
					case TEAM_NEXT:// 取下一缓存块
						if (tmp.getOffset_file() == rafOffset) {
							rafOffset = tmp.getEnd();
							fbt = albt.remove(i);
							return fbt;
						}
						break;
					case LATELY:// 取就近缓存快
						if (fbt != null && tmp.getOffset_file() < fbt.getOffset_file()) {
							fbt = tmp;
						}
						break;
					}
				}
			}
	
			// 缓存就近输出模式
			if (type == LATELY && fbt != null && albt.size() > 0) {
				albt.remove(fbt);
			}
		}
		return fbt;
	}

	public long getRemnantByteCount() {
		return length - achieve;
	}

	public long getRemnantTime() {
		return getRemnantByteCount() / (Speed == 0 ? 1 : Speed);
	}

	public XML getSaveXML() {
		/**
		 * <4a48da31e9bd0cf7 achieve=4408 size=4480 time=1321064
		 * filename="\0.jpg" > <task offset=0 end=655 address="127.0.0.1"
		 * port=1346 UserID=* /> <TASK offset=655 end=65545
		 * address="192.168.0.0.1" port=1346 UserID=2/> <WAIT
		 * address="192.168.0.0.1" port=1346 UserID=2/> </4a48da31e9bd0cf7>
		 * 
		 */
		xml.FLAG = md5;

		xml.put("filepath", getF().getAbsolutePath());
		xml.put("size", String.valueOf(length));
		xml.put("time", String.valueOf(LastModified));

		xml.put("@transmitTime", String.valueOf(transmitTime));
		xml.put("@achieve", String.valueOf(achieve));
		xml.put("@ThreadSum", String.valueOf(threadSum));
		xml.put("@rafOffset", String.valueOf(rafOffset));
		xml.put("@alEnd", String.valueOf(alEnd));
		xml.put("@end", String.valueOf(end));

		XMLNode xfs = new XMLNode(null, md5);
		xfs.copy(xml);

		xml.clear();
		xml.add("", xfs, XML.TOP);

		for (int i = 0; i < altask.size(); i++) {
			PipThread fs = altask.get(i);
			XMLNode xf = new XMLNode(null, "TASK");
			xf.put("gcode", fs.getGcode());
			xf.put("gucode", fs.getGucode());
			xf.put("userid", fs.getUserID());
			if (fs.getAddress() != null) {
				xf.put("address", fs.getAddress().getHostAddress());
				xf.put("port", String.valueOf(fs.getPort()));
			}
			xf.put("offset", String.valueOf(fs.offset));
			xf.put("end", String.valueOf(fs.end));
			xf.put("over", fs.over);
			xml.add(md5, xf, XML.BUTTOM);
		}
		for (int i = 0; i < alwait.size(); i++) {
			PipThread fs = alwait.get(i);
			if (fs != null) {
				XMLNode xf = new XMLNode(null, "WAIT");
				xf.put("userid", fs.getUserID());
				if (fs.getAddress() != null) {
					xf.put("address", fs.getAddress().getHostAddress());
					xf.put("port", String.valueOf(fs.getPort()));
				}
				xf.put("offset", String.valueOf(fs.offset));
				xf.put("end", String.valueOf(fs.end));
				xml.add(md5, xf, XML.BUTTOM);
			}
		}
		// System.out.println("[任务配置文件]"+xml);
		xml.createCode(XMLFile);
		return xml;
	}

	public ShareRoot getSharePath() {
		return sp;
	}

	/**
	 * 取文件大小（浓缩单位）
	 * 
	 * @return
	 */
	public String getSizeStr() {
		return Units.byteCondense(this.getF().length());
	}

	/**
	 * 缓存输出速率
	 * 
	 * @return
	 */
	public long getSpeed() {
		return Speed;
	}

	public int getTaskSum() {
		return altask.size();
	}

	public LinkedList<PipThread> getWaitingPips() {
		return alwait;
	}

	public int getWaitSum() {
		return alwait.size();
	}

	public XML getXML() {
		return xml;
	}

	public File getXmlFile() {
		return XMLFile;
	}

	/**
	 * 判断是否有刚连接的用户
	 * 
	 * @return
	 */
	public boolean haveConnected() {
		PipThread fs = null;
		for (int i = 0; i < alwait.size(); i++) {
			fs = alwait.get(i);
			if (fs != null && fs.speed == -1) {
				return true;
			}
		}
		return false;
	}

	/*
	 * RandomAccessFile 随机访问文件流 "r" 以只读方式打开。调用结果对象的任何 write 方法都将导致抛出
	 * IOException。 "rw" 打开以便读取和写入。如果该文件尚不存在，则尝试创建该文件。 "rws" 打开以便读取和写入，对于
	 * "rw"，还要求对文件的内容或元数据的每个更新都同步写入到底层存储设备。 "rwd"
	 * 打开以便读取和写入，对于"rw"，还要求对文件内容的每个更新都同步写入到底层存储设备。
	 */

	/**
	 * 触发事件
	 * 
	 * @param type
	 */
	private void infomEvent(int type) {
		for (int i = 0; i < listener.size(); i++) {
			switch (type) {
			case EVENT_ORVER:
				listener.get(i).achieveEvent(this);
				break;
			case EVENT_PERIODS:
				listener.get(i).periodsEvent(this);
				break;
			case EVENT_EXCEPTION:
				listener.get(i).TaskExceptionEvent(this);
				break;
			case EVENT_PAUSE:
				listener.get(i).PauseEvent(this);
				break;
			}
		}
	}
	public boolean isDoing() {
		return doing;
	}
	public boolean isOK() {
		return achieve >= length;
	}

	/**
	 * 从待命区取出一个优秀种子
	 * 
	 * @param speed
	 * @return null表示没有待命种子
	 */
	public PipThread nextPip() {
		if (alwait.size() == 0)
			return null;
		int max = 0;
		PipThread good = null;
		boolean hasNewConn = haveConnected();
		synchronized (alwait) {
			good = alwait.get(max);
			for (int i = 1; i < alwait.size(); i++) {
				PipThread fs = alwait.get(i);
				// 刚刚连接进入的优先级最高
				if (fs != null && fs.speed == -1 ? fs.conConsume < good.conConsume : fs.getSpeedNet() > good.getSpeedNet()) {
					good = fs;
				}
			}
			alwait.remove(good);
		}
		return good;
	}
	/**
	 * 暂停任务
	 */
	public void pause() {
		pause(true);
	}
	/**
	 * 暂停任务
	 * 
	 * @param b
	 *            是否触发事件
	 */
	public void pause(boolean b) {
		if (doing) {
			doing = false;
			timer.cancel();// 清除时钟任务
			try {
				Thread.sleep(500);
			} catch (Exception e) {
			}
			timer.purge();// 移出已取消任务
			if (b)
				infomEvent(EVENT_PAUSE);
		}
	}
	/**
	 * 判断种子是否存在
	 * 
	 * @param p
	 * @return
	 */
	public boolean pipExist(Pip p) {
		PipThread fa;
		// 查找待命区是否存在
		for (int i = 0; i < alwait.size(); i++) {
			fa = alwait.get(i);
			if (p.equals(fa))
				return true;
		}

		for (int i = 0; i < altask.size(); i++) {
			fa = altask.get(i);
			if (fa.equals(p) && !fa.Stop)// 判断地址和端口是否相等
				return true;
		}
		return false;
	}
	public void removeListener(DownloadTaskListener dtl) {
		listener.remove(dtl);
	}
	public void reset() {
		altask.clear();
		alwait.clear();
		albt.clear();
		Speed = 0;
		doing = false;
		achieve = 0;
		rafOffset = 0;
		alEnd = 0;
		end = 0;
		transmitTime = 1;

	}
	/**
	 * 缓存输出线程（持久）
	 */
	@Override
	public void run() {
		try {
			tmpfile = new File(this.getF().getAbsolutePath()+".tmp");
			raf = new RandomAccessFile(tmpfile, "rw");
			raf.setLength(length);
			Log.print("Cache write start "+tmpfile);
			end = 0;
			ParcelBytes fbt = null;
			long offset;
			int size;
			while (!isOK() && output == Thread.currentThread()) {
//				Log.print("Cache list size: "+albt.size()+"  "+getF());
				while ((fbt = getNextFB(wise)) != null) {
					offset = fbt.getOffset_file();
					size = fbt.getSize();

					raf.seek(offset); // 文件指针偏移
					raf.write(fbt.getBytes(), 0, size); // 输出

					achieve += size; // 统计已完成字节数

					end = fbt.getEnd(); // 设置最后写入位置

					fbt.reclaim(); // 回收字节块
					Log.print("Cache write to tmp-file:"+tmpfile+"  size:"+size+"  md5:"+md5);
				}
				try {
					Units.sleep(1000);
				} catch (Exception e) {
				} // 让出CPU资源
			}
			raf.close();
			tmpfile.setLastModified(LastModified);// 设置最后修改时间
			getF().delete();
			tmpfile.renameTo(getF());
			Log.print("Cache write over: "+tmpfile);
		} catch (Exception e) {
			if (doing)
				Log.print("[缓存线程]输出异常:" + e.getMessage());
			e.printStackTrace();
		} finally {
			try {
				raf.close();
			} catch (Exception e) {
			}
			try {
				Thread.sleep(1000);
			} catch (InterruptedException e) {
			}
			// this.open();//自动打开关联文件
			getSaveXML();
			pause(false);// 停止任务
			if (isOK()) {
				infomEvent(EVENT_ORVER);// 触发完成事件
			}
			output = null;// 线程结束清空
		}
	}
	/**
	 * 启动任务
	 */
	public void start() {
		if (!doing) {
			doing = true;
			MainWindows.getInstance().client.send("<get md5=\"" + this.md5 + "\" />");

			// 启动写缓存线程
			if (output == null) {
				output = new Thread(this);
				output.start();
			}
			tt = new TimerTask() {

				@Override
				public void run() {
					if (!doing)
						return;
					transmitTime += 500;// 任务累计运行时
					Speed = (achieve / transmitTime) * 1000;

					// 处理待命区，挑选优秀连接进行分配任务
					allot();

					/***********************************************************
					 * 触发周期事件
					 **********************************************************/
					infomEvent(EVENT_PERIODS);

				}
			};
			timer = new Timer();
			timer.schedule(tt, 500, 500);// 种子任务分配和显示处理
		}
	}

	public File getTmpfile(){
		return tmpfile;
	}
	
	@Override
	public String toString() {
		return "[DownLoadTask: MaxThreadSum=" + threadSum + "]";
	}

}
